package sicnu.cs.ich.rocketmq.rocketmq;

import lombok.Data;
import org.apache.logging.log4j.util.Strings;
import org.apache.rocketmq.client.consumer.MQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListener;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Arrays;

/**
 * @author CaiKe
 * @date 2021/12/1 23:13
 */
@Data
public abstract class AbstractRocketConsumer implements RocketConsumer {

    public static final String METHOD_PREFIX = "consumeTag";

    protected String topic;
    protected String tag;
    protected String consumerTitle;
    private MessageListenerConcurrently listener;
    protected MQPushConsumer pushConsumer;

    /**
     * 必要信息
     *
     * @param topic
     * @param tag
     * @param consumerTitle
     */
    public void necessary(String topic, String tag, String consumerTitle) {
        this.topic = topic;
        this.tag = tag;
        this.consumerTitle = consumerTitle;
    }

    @Override
    public void registerMessageListener(MessageListenerConcurrently listener) {
        this.listener = listener;
    }


    /**
     * 处理消息
     */
    protected void handleMessage(){
        registerMessageListener((msgs, context) -> {
            for (MessageExt msg : msgs) {
                try {
                    return consumeMsg(msg, msg.getTags());
                } catch (Exception e) {
                    e.printStackTrace();
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
    }

    /**
     * 通过反射调用方法名处理消息
     * @param message
     * @param tag
     * @return
     * @throws NoSuchMethodException
     * @throws InvocationTargetException
     * @throws IllegalAccessException
     */
    protected ConsumeConcurrentlyStatus consumeMsg(MessageExt message, String tag) throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
        if (Strings.isBlank(tag)){
            tag = "default";
        }
        Method method;
        try {
            method = this.getClass().getDeclaredMethod(tag, MessageExt.class);
        } catch (NoSuchMethodException e) {
//            e.printStackTrace();
            throw new NoSuchMethodException("无该方法：" + tag + "，" + e.getMessage());
        }
        return (ConsumeConcurrentlyStatus) method.invoke(this, message);
    }
}